package com.example.cdc;

import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import com.ververica.cdc.connectors.mysql.table.StartupOptions;
import com.ververica.cdc.debezium.StringDebeziumDeserializationSchema;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * Created with IntelliJ IDEA.
 * ClassName: FlinkCDC
 * Package: com.example.cdc
 * Description:
 * User: fzykd
 *
 * @Author: LQH
 * Date: 2023-08-08
 * Time: 11:33
 */

public class FlinkCDC {

    public static void main(String[] args) throws Exception {

        //1获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);

        //通过FlinkCDC 构造SourceFunction
        MySqlSource<String> sourceFunction = MySqlSource.<String>builder()
                //添加连接的参数
                .hostname("hadoop102")
                .port(3306)
                .username("root")
                .password("000000")
                //指定要监听的数据库
                .databaseList("cdc_test")
                //哪个数据表 如果不用添加参数 那么就是监听库下面所以的表
                .tableList("cdc_test.user_info")
                //反序列化器
                .deserializer(new StringDebeziumDeserializationSchema())
                .startupOptions(StartupOptions.initial())
                .build();

        DataStreamSource<String> stringDataStreamSource = env.fromSource(sourceFunction,
                WatermarkStrategy.noWatermarks(), "MySQL Source");


        stringDataStreamSource.print();
        env.execute("flinkCDC");


    }

}
